package com.ibm.cps.processors;

import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

import javax.mail.Authenticator;
import javax.mail.Message;
import javax.mail.PasswordAuthentication;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;

import org.apache.log4j.Logger;

import com.google.common.base.Throwables;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.SchemaFormat;
import com.ibm.factories.MessageParserFactory;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;
// need to extends
public class EmailPublishProcessor extends BasicProcessor {

	@ValueParam
	private String destinationMailAddress;
	private ITimeSeriesSchema parentSchema;
	private Session session;
	private static Logger logger = Logger.getLogger(EmailPublishProcessor.class);
	private IMessageParser parser; 

	public EmailPublishProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
		
		/* TODO here I hard code email to use json for 
		 * simplicity. may need change */
		this.parser = MessageParserFactory.createMessageParser(SchemaFormat.json, parentSchema);
		initSession();
	}

	private void initSession() {
		System.setProperty("java.net.preferIPv4Stack", "true");
		// Get system properties
		Properties props = new Properties();

		// Setup mail server
		props.put("mail.smtp.host", LocalConfig.MAIL_SMTP_HOST);
		props.put("mail.smtp.port", String.valueOf(LocalConfig.MAIL_SMTP_PORT));
		props.put("mail.smtp.auth", "true");
		props.put("mail.smtp.starttls.enable", "true");
		props.put("mail.smtp.socketFactory.port", "465");
		props.put("mail.smtp.socketFactory.class",
				"javax.net.ssl.SSLSocketFactory");

		session = Session.getDefaultInstance(props, new Authenticator() {
			protected PasswordAuthentication getPasswordAuthentication() {
				return new PasswordAuthentication(LocalConfig.FROMEMAIL,
						LocalConfig.FROMEMAILPASSWORD);
			}
		});
		session.setDebug(true);
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message)
			throws CPSException {
		ParsedDataMessage parsedMessage = (ParsedDataMessage) message;
		if (parsedMessage.getValueCount() != parentSchema.getTagCount()) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"The count of values( "
							+ Arrays.toString(parsedMessage.getObjectValues())
							+ ") should be equal to schema tag count ("
							+ Arrays.toString(parentSchema.getTagNames())
							+ ").");
		}
		try {
			// Define message
			MimeMessage mailMessage = new MimeMessage(session);
			mailMessage.setFrom(new InternetAddress(LocalConfig.FROMEMAIL));
			mailMessage.addRecipient(Message.RecipientType.TO,
					new InternetAddress(destinationMailAddress));

			mailMessage.setSubject("Alert from realtime actional system.");

			if (parsedMessage.getObjectValues() != null) {
				
				byte[] retMsg = parser
						.restoreFromDataMessage(parsedMessage);
				mailMessage.setText(new String(retMsg));
			}
			mailMessage.saveChanges();

			Transport.send(mailMessage);
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Can't send alert mail to " + destinationMailAddress);
		}
		return null;
	}

	@Override
	public void dispose() {

	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() {
		return null;
	}

	@Override
	public void open() throws CPSException {
		
	}

}